package ntd.block.mqtt;


import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.log.StaticLog;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * mqtt client
 */
public class ComMqttClient {

    public MqttClient client;

    private Map<String, IMqttMessageListener> listenHashMap = new HashMap<>();

    public ComMqttClient(String url, String username, String password) {
        if (!StrUtil.isAllNotBlank(url, username, password)) {
            throw new RuntimeException("mqtt 初始化失败,参数传入异常...");
        }
        ComMqttClient comMqttClient = this;
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            client = new MqttClient(url, "com-server-client-" + RandomUtil.randomString(8), persistence);
            // MQTT 连接选项
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setUserName(username);
            connOpts.setPassword(password.toCharArray());
            // 保留会话
            connOpts.setCleanSession(true);
            // 自动重连
            connOpts.setAutomaticReconnect(true);
            // 设置客户端可以同时发送的消息数量为1000，默认为10，超过会提示《32202：正在进行过多的发布》
            connOpts.setMaxInflight(1000);
            // 设置回调
            client.setCallback(new MqttCallback() {
                public void connectionLost(Throwable cause) {
                    // 连接丢失后，一般在这里面进行重连
                    while (!client.isConnected()) {
                        // 每隔10秒重试一次
                        try {
                            TimeUnit.SECONDS.sleep(10);
                            StaticLog.info("mqtt - 尝试重连......");
                            if (!client.isConnected()) {
                                client.reconnect();
                            } else {
                                break;
                            }
                        } catch (MqttException e) {
                            e.printStackTrace();
                            StaticLog.info("mqtt - 重连失败......");
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    StaticLog.info("mqtt - 重连成功......");
                    // 重新订阅主题
                    listenHashMap.entrySet().forEach(listenerEntry -> comMqttClient.subscribe(listenerEntry.getKey(), listenerEntry.getValue()));
                }

                public void messageArrived(String topic, MqttMessage message) {
                    // subscribe后得到的消息会执行到这里面
//                    StaticLog.info("接收MQTT消息[{}][Qos:{}][{}]", topic, message.getQos(), new String(message.getPayload()));
                }

                public void deliveryComplete(IMqttDeliveryToken token) {
//                    System.out.println("deliveryComplete---------" + token.isComplete());
                }
            });
            client.connect(connOpts);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) {
        ComMqttClient comMqttClient = new ComMqttClient("tcp://61.52.247.249:12020", "com_server", "com_server");

        comMqttClient.subscribe("test/#", (topic, message) -> System.out.println("接收消息内容:" + new String(message.getPayload())));

        for (int i = 0; i < 5; i++) {
            comMqttClient.pushQos1("/test/1", "测试数据！！！！！！！！！！！！！！！！！！！" + i);
        }

//        comMqttClient.close();
    }

    /**
     * 订阅消息
     *
     * @param topic
     * @param iMqttMessageListener
     */
    public void subscribe(String topic, IMqttMessageListener iMqttMessageListener) {
        try {
            client.subscribe(topic, 2, iMqttMessageListener);
            listenHashMap.put(topic, iMqttMessageListener);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 推送消息
     *
     * @param topic
     * @param msg
     * @param qos
     */
    public void push(String topic, String msg, int qos) {
        try {
            MqttMessage message = new MqttMessage(msg.getBytes());
            message.setQos(qos);
            client.publish(topic, message);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    public void pushQos0(String topic, String msg) {
        push(topic, msg, 0);
    }

    public void pushQos1(String topic, String msg) {
        push(topic, msg, 1);
    }

    public void pushQos2(String topic, String msg) {
        push(topic, msg, 2);
    }

    /**
     * 关闭连接
     */
    public void close() {
        try {
            client.disconnect();
            client.close();
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
}
